package com.atguigu.gmallcanal;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.atguigu.gmall.constances.CommonConstants;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Random;

public class CanalClient {
    /*① 使用当前客户端订阅canalserver
            a)创建连接
                    集群连接模式:支持failover,借助了zk,要想使用得去配置文件配zk
                    普通连接
                            参数:   SocketAddress address :canal服务端的主机名和端口号构造的socket
                                    String destination:订阅哪个mysql实例
                                    String username:用户名
                                    String password:密码
      ② 订阅gmallrealtime的order_info表
      ③ get订阅数据,按批次订阅,解析数据,将order_info新增的订单的每一列列名和列值都获取到,封装为json

            message:一次拉取的一批数据,多条SQL
                List<Entry> entrys:记录拉取的数据中,每一个SQL所影响的行的变化

                Entry:
                    header:记录一些属性或元数据例如tableName
                    entryType:记录语句的类型,是开启事务还是增删改查(RowData)
                    StoreValue:记录entry影响的数据变化(序列化数据,需要反序列化)

            工具类: RowChange
                RowChange可以将StoreValue转为一个RowChange对象
                        EventType:记录语句的类型 insert delete
                        RowDataList: 记录语句影响的每一行
                                RowData : 影响的一行
                                    column: 一行的一列
                                        列名,列之前的值和之后的值都能获取到
      ③ 将json写入kafka

    * */
    public static void main(String[] args) throws InvalidProtocolBufferException {

        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("hadoop102", 11111), "example", "", "");

        canalConnector.connect();
        canalConnector.subscribe("gmall_realtime.*");

        //流式数据,源源不断拉取
        while (true) {

            Message message = canalConnector.get(100);
            //如果没有拉取到数据
            if (message.getId() == -1) {
                try {
                    Thread.sleep(1000);
                    System.out.println("没有数据,我歇会");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                continue;
            }
            //如果拉取到了数据,进行解析
            List<CanalEntry.Entry> entries = message.getEntries();
            System.out.println(message);
            //拿到每一个entry
            for (CanalEntry.Entry entry : entries) {
                if (entry.getEntryType().equals(CanalEntry.EntryType.ROWDATA)) {
                    String tableName = entry.getHeader().getTableName();
                    ByteString storeValue = entry.getStoreValue();
                    System.out.println(tableName);
                    //解析storevalue
                    parseStoreValue(tableName, storeValue);
                }
            }
        }
    }

    private static void parseStoreValue(String tableName, ByteString storeValue) throws InvalidProtocolBufferException {
        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);

        if ("order_info".equals(tableName) && rowChange.getEventType().equals(CanalEntry.EventType.INSERT) ) {
            System.out.println("写入kafka主题order_info");
            writeToKafka(rowChange, CommonConstants.KAFKA_ORDER_INFO);
        } else if ("order_detail".equals(tableName) && rowChange.getEventType().equals(CanalEntry.EventType.INSERT)) {
            System.out.println("写入kafka主题order_detail");
            writeToKafka(rowChange, CommonConstants.KAFKA_ORDER_DETAIL);
        } else if ("user_info".equals(tableName) && (rowChange.getEventType().equals(CanalEntry.EventType.INSERT)
                || rowChange.getEventType().equals(CanalEntry.EventType.UPDATE))) {
            System.out.println("写入kafka主题user_info");
            writeToKafka(rowChange, CommonConstants.KAFKA_USER_INFO);
        }
    }

    public static void writeToKafka(CanalEntry.RowChange rowChange, String topic) {
        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
        //遍历insert语句影响的每一行
        for (CanalEntry.RowData rowData : rowDatasList) {

            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();

            JSONObject jsonObject = new JSONObject();
            //遍历一行的列
            for (CanalEntry.Column column : afterColumnsList) {

                jsonObject.put(column.getName(), column.getValue());
            }
//            //在每条数据写往kafka前,模拟canal网络延迟导致数据无法同时传输给kafka的情况
//            int delayTime = new Random().nextInt(10);//延迟随机秒发送kafka
//            try {
//                Thread.sleep(delayTime*1000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
            //打印要在遍历一行每列的外面
            MyProducer.writeToKafka(topic, jsonObject.toJSONString());
        }
    }
}

